package org.baojie.raft.delay;

import org.baojie.raft.util.current.LocalSleep;
import org.baojie.raft.util.current.RaftFactory;
import org.baojie.raft.util.current.RaftThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class MessageCache<K, V extends AbstractValue> implements RaftCache<K, V> {

    private static final Logger log = LoggerFactory.getLogger(MessageCache.class);

    private final ConcurrentLinkedQueue<DelayKey<K>> remove = new ConcurrentLinkedQueue<>();
    private final ArrayList<Future<?>> futures = new ArrayList<>(32);
    private final AtomicBoolean stop = new AtomicBoolean(false);
    private final DelayQueue<DelayKey<K>> delays = new DelayQueue<>();
    private final ReentrantLock mainLock = new ReentrantLock();

    private final int contains;
    private final RaftThreadPool workers;
    private final AtomicInteger size = new AtomicInteger(0);
    private final ConcurrentHashMap<K, V> cache;

    private MessageCache(int size) {
        this.contains = size;
        this.workers = workers();
        this.cache = new ConcurrentHashMap<>(size);
        prepare();
    }

    private RaftThreadPool workers() {
        return new RaftThreadPool(4,
                8,
                180,
                TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                RaftFactory.create("cache for message", false));
    }

    private void prepare() {
        delayWork();
        removeWork();
    }

    private void delayWork() {
        for (int i = 0; i < 2; i++) {
            Delayer delayer = new Delayer();
            Future<?> future = workers.submit(delayer);
            futures.add(future);
        }
    }

    private void removeWork() {
        for (int i = 0; i < 4; i++) {
            Remover remover = new Remover();
            Future<?> future = workers.submit(remover);
            futures.add(future);
        }
    }

    public static final MessageCache create(int size) {
        if (size <= 0) {
            throw new IllegalArgumentException();
        } else {
            return new MessageCache(size);
        }
    }

    @Override
    public List<K> touchKeys() {
        List<K> keys = new ArrayList<>(cache.size());
        Iterator<K> iterator = cache.keySet().iterator();
        while (iterator.hasNext()) {
            K k = iterator.next();
            keys.add(k);
        }
        return keys;
    }

    @Override
    public V get(K k) {
        if (null == k) {
            return null;
        } else {
            return cache.get(k);
        }
    }

    @Override
    public V putIfAbsent(K k, V v, long keep, TimeUnit unit) {
        if (null == k) {
            return null;
        }
        if (null == v) {
            return null;
        }
        if (keep <= 0) {
            return null;
        }
        if (stop.get()) {
            return null;
        }
        if (size.get() >= contains) {
            return null;
        } else {
            return realPut(k, v, keep, unit);
        }
    }

    private V realPut(K k, V v, long keep, TimeUnit unit) {
        V temp = cache.putIfAbsent(k, v);
        if (null == temp) {
            DelayKey<K> key = new DelayKey<>(k, keep, unit);
            try {
                delays.offer(key);
            } finally {
                size.incrementAndGet();
            }
        }
        return temp;
    }

    @Override
    public final int contains() {
        return contains;
    }

    @Override
    public final int size() {
        return size.get();
    }

    @Override
    public V remove(K k) {
        if (null == k) {
            return null;
        }
        V temp = cache.remove(k);
        if (null != temp) {
            try {
                size.decrementAndGet();
            } finally {
                // 这里会出现缓存抖动,
                // 同样的key
                // 原来的key已经过期被删除线程删除删除
                // 然后马上又添加了相同key的新值
                // 但是这时执行了当前的remove操作
                // 使得缓存出现抖动
                // 对缓存进行单一线程使用减少此问题
                // delays.remove(k);
            }
        }
        return temp;
    }

    @Override
    public final boolean close() {
        if (!stop.get()) {
            if (stop.compareAndSet(false, true)) {
                cancelFutures();
                shutDownPool();
                clean();
                return true;
            }
        }
        return false;
    }

    @Override
    public final boolean clean() {
        if (stop.get()) {
            return false;
        } else {
            final ReentrantLock lock = mainLock;
            lock.lock();
            try {
                cache.clear();
                delays.clear();
                size.set(0);
                return true;
            } finally {
                lock.unlock();
            }
        }
    }

    private final void cancelFutures() {
        for (Future f : futures) {
            f.cancel(true);
        }
    }

    private final void shutDownPool() {
        for (int i = 0; i < 100; i++) {
            workers.shutdown();
            if (workers.isTerminated()) {
                break;
            } else {
                sleep(30, TimeUnit.MILLISECONDS);
            }
        }
        if (!workers.isTerminated()) {
            workers.shutdownNow();
        }
    }

    private final void sleep(long sleep, TimeUnit unit) {
        Thread.yield();
        LocalSleep.park(sleep, unit);
        Thread.yield();
    }

    private final class Delayer implements Runnable {

        public Delayer() {

        }

        @Override
        public void run() {
            DelayKey<K> key = null;
            while (!stop.get()) {
                key = getKeys();
                if (null == key) {
                    continue;
                } else {
                    toRemove(key);
                }
            }
        }

        private DelayKey<K> getKeys() {
            try {
                return delays.take();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            }
            return null;
        }

        private void toRemove(DelayKey<K> key) {
            try {
                remove.offer(key);
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            } finally {
                K k = key.getKey();
                if (null != k) {
                    if (null != cache.remove(k)) {
                        size.decrementAndGet();
                    }
                }
            }
        }

    }

    private final class Remover implements Runnable {

        public Remover() {

        }

        @Override
        public void run() {
            DelayKey<K> key = null;
            while (!stop.get()) {
                key = pollKeys();
                if (null == key) {
                    sleep(100, TimeUnit.MILLISECONDS);
                    checkElems();
                } else {
                    K k = key.getKey();
                    if (null != k) {
                        remove(k);
                    } else {
                        log.error("get K from Key null, key=" + key);
                    }
                }
            }
        }

        private DelayKey<K> pollKeys() {
            try {
                return remove.poll();
            } catch (Throwable t) {
                log.error(t.getMessage(), t);
            }
            return null;
        }

        private void remove(K k) {
            if (null != cache.remove(k)) {
                size.decrementAndGet();
            }
        }

        private void checkElems() {
            int offset = contains / 3;
            if (0 == offset) {
                offset = 3;
            }
            int s = cache.size();
            int as = size.get();
            if (s >= contains + offset) {
                log.warn("cache size=" + s + ", contains=" + contains + ", atomic size=" + as);
            }
        }

    }

}